package com.dianmi.seckill.service.impl;

import com.dianmi.seckill.common.advice.ResponseCode;
import com.dianmi.seckill.common.advice.ServiceException;
import com.dianmi.seckill.common.utils.AssertUtil;
import com.dianmi.seckill.common.utils.FastUtils;
import com.dianmi.seckill.common.utils.JsonUtil;
import com.dianmi.seckill.config.RocketMQConstant;
import com.dianmi.seckill.entity.SeckillProduct;
import com.dianmi.seckill.entity.SubstractStockInfo;
import com.dianmi.seckill.service.SeckillService;
import com.dianmi.seckill.service.StockCoordinate;
import com.dianmi.seckill.vo.SeckillInfo;
import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;

/**
 * @author xieqiaofu
 * @date 2021/4/7
 */
@Slf4j
@Service
public class SeckillServiceImpl implements SeckillService {

    private static final String USER_FLAG_LUA =
            "if redis.call(\"exists\", KEYS[1]) == 1 " +
            "then " +
            "   return tonumber(redis.call(\"get\", KEYS[1])) " +
            "else " +
            "   redis.call(\"set\", KEYS[1], ARGV[1]) " +
            "   return 0 " +
            "end ";

    private static final String HAS_JOIN = "请勿重复参与";

    private static final String HAS_SOLD_OUT = "商品已卖完";

    private static final String QUEUING = "排队中";

    private static final String SECKILL_ERROR = "秒杀失败";

    private static final String USER_FLAG_KEY = "user_flag_prefix:%s:%s";

    private static final String SEND_MQ_ERROR_MSG = "send product: %s to mq failed";

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private ProductServiceImpl productService;

    @Autowired
    private StockCoordinate stockCoordinate;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    private StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

    @Override
    public String seckill(SeckillInfo seckillInfo) {
        /*1、判断剩余库存标志*/
        AssertUtil.allow(productService.stockFlag(seckillInfo.getProductId()), HAS_SOLD_OUT);
        /*2、查询或生成用户标记*/
        generateUserFlag(seckillInfo);
        /*3、扣减库存*/
        Boolean reduce = productService.reduce(seckillInfo.getProductId());
        if (!reduce) {
            //修改本地商品库存标志
            productService.updateStockFlag(seckillInfo.getProductId(), Boolean.FALSE);
            //通知其他节点修改商品库存标志
            stockCoordinate.update(seckillInfo.getProductId(), false);
            throw new ServiceException(500, HAS_SOLD_OUT);
        }
        boolean seckillSucc = true;
        try {
            SeckillProduct seckillProduct = productService.getProductInfo(seckillInfo.getProductId());
            AssertUtil.allow(seckillProduct != null, ResponseCode.SERVICE_ERROR.getMsg());

            SubstractStockInfo substractStockInfo = FastUtils.convertCopy(SubstractStockInfo.class, seckillProduct);
            substractStockInfo.setUserId(seckillInfo.getUserId());
            Message message = new Message(RocketMQConstant.SECKILL_PRODUCT_STOCK_TOPIC, JsonUtil.toJson(substractStockInfo).getBytes(StandardCharsets.UTF_8));
            rocketMQTemplate.asyncSend(RocketMQConstant.SECKILL_PRODUCT_STOCK_TOPIC, message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    if (SendStatus.SEND_OK == sendResult.getSendStatus()) {
                        String userFlagKey = userFlagKey(seckillInfo.getProductId(), seckillInfo.getUserId());
                        //修改用户标记为排队中
                        redisTemplate.opsForValue().set(userFlagKey, String.valueOf(2));
                    }
                }

                @Override
                public void onException(Throwable throwable) {
                    handleSendFailed(reduce, seckillProduct.getId(), seckillInfo.getUserId(), throwable);
                }
            });
        } catch (Exception e) {
            seckillSucc = false;
            handleSendFailed(reduce, seckillInfo.getProductId(), seckillInfo.getUserId(), e);
        }
        return seckillSucc ? QUEUING : SECKILL_ERROR;
    }

    private void handleSendFailed(Boolean reduce, Long productId, Long userId, Throwable throwable) {
        String errMsg = String.format(SEND_MQ_ERROR_MSG, productId);
        log.error(errMsg, throwable);
        //还原redis库存
        if (reduce) {
            productService.restore(productId);
        }
        //清除用户标记，让用户可以再次秒杀。如果不清除，可能没有人能够抢到
        redisTemplate.delete(userFlagKey(productId, userId));
    }

    private void generateUserFlag(SeckillInfo seckillInfo) {
        String userFlagKey = userFlagKey(seckillInfo.getProductId(), seckillInfo.getUserId());
        RedisScript<Long> script = new DefaultRedisScript<>(USER_FLAG_LUA, Long.class);
        Object execute = redisTemplate.execute(script, stringRedisSerializer, stringRedisSerializer, Lists.newArrayList(userFlagKey), "1");
        Long ret = (Long)execute;
        if (ret != null && ret.equals(3L)) {
            throw new ServiceException(500, HAS_JOIN);
        } else if (ret != null && ret.equals(2L)) {
            throw new ServiceException(500, QUEUING);
        } else if (ret != null && ret.equals(1L)) {
            throw new ServiceException(500, SECKILL_ERROR);
        }
    }

    public void userJoinFlag(Long productId, Long userId) {
        redisTemplate.opsForValue().set(userFlagKey(productId, userId), String.valueOf(3));
    }

    public Long queryUserFlag(Long productId, Long userId) {
        return (Long) redisTemplate.opsForValue().get(userFlagKey(productId, userId));
    }

    private String userFlagKey(Long productId, Long userId) {
        return String.format(USER_FLAG_KEY, productId, userId);
    }
}
